package services

import (
	"context"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"longmen/server/config/global"
	global2 "longmen/server/config/global"
	"longmen/server/internal/rpc/tencentim/imsdk"
	publisher2 "longmen/server/internal/services/consumers/publisher"
	_const "longmen/server/pkg/common/helper/const"
	"longmen/server/pkg/common/helper/encrypt"
	string2 "longmen/server/pkg/common/helper/string"
	"longmen/server/pkg/common/helper/timehelper"
	"longmen/server/pkg/common/helper/util"
	"longmen/server/pkg/common/sdk/tx"
	"longmen/server/pkg/db/mysql"
	"longmen/server/pkg/db/mysql/db_live"
	"longmen/server/pkg/db/mysql/db_sports"
	"longmen/server/pkg/db/mysql/models/models_guess"
	"longmen/server/pkg/db/mysql/models/models_live"
	"longmen/server/pkg/db/mysql/models/models_main"
	"longmen/server/pkg/db/redis/cache"
	pt "longmen/server/pkg/protobuf"
	"net"
	"strconv"
	"strings"
	"time"

	"github.com/godruoyi/go-snowflake"
	"github.com/olivere/elastic/v7"

	"github.com/spf13/cast"

	"longmen/server/pkg/common/api_models"

	"github.com/jinzhu/copier"
)

func (s *Service) CheckIsMnanager(anchorid int, user *models_main.User) (bool, error) {
	fName := "CheckIsMnanager"
	//判断該用戶是否有權限禁言/解禁
	managerUser, err := db_live.GetLiveRoomManagerParam(map[string]interface{}{"mgrid": user.Id})
	if err != nil {
		return false, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}
	if managerUser.Id > 1 {
		if managerUser.Anchorid == 0 || managerUser.Anchorid == anchorid { //不是超管和该房间管理
			return true, nil
		}
	}
	return false, nil
}

func (s *Service) GetBaseCategory(channel string, platform int8) ([]*models_live.LiveCategory, error) {
	resp, err := cache.GetLiveCategory()
	if err != nil {
		resp, err = db_live.GetLiveCategoryParam(map[string]interface{}{"status": 1}, "sort, id")
		_ = cache.SetLiveCategory(resp)
	}
	return resp, err

	//result := make([]*models_live.LiveCategory, 0)
	//if channel != "" && channel != _const.LONGMEN_ANDROID_APP_CHANNEL && platform == _const.LOGIN_PLATFORM_ANDROID {
	//	for _, val := range resp {
	//		if val.Category != _const.LiveCategoryReplayType {
	//			result = append(result, val)
	//		}
	//	}
	//} else {
	//	result = resp
	//}
	//return result, err
}

func (s *Service) EditLive(r *api_models.EditLiveReq, user *models_main.User) (string, *models_live.Live, error) {
	fName := "EditLive"
	live, err := db_live.GetLiveParam(map[string]interface{}{"liveid": r.Liveid, "anchorid": user.Id})
	if err != nil {
		return "", nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}

	if live.Anchorid <= 0 {
		return "直播不存在", nil, fmt.Errorf("%s 直播不存在", util.Convert(fName))
	}

	live.Title = r.Title
	live.Thumb = r.Thumb
	live.Categoryid = r.Cateid
	live.Matchid = r.MatchId
	live.SportType = r.SportType
	live.Ticket = r.Ticket
	live.EnableWebrtc = r.EnableWebrtc

	liveCol := []string{"title", "thumb", "categoryid", "matchid", "sport_type", "ticket", "enable_webrtc"}
	if live.Categoryid == 1 || live.Categoryid == 2 || live.Categoryid == 3 || live.Categoryid == 11 {
		live.PullUrl = r.PullUrl
		liveCol = append(liveCol, "pull_url")
	}
	err = db_live.UpdateLiveByTransactions(nil, liveCol, live)
	if err != nil {
		return "", nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}
	return "", live, nil
}

func (s *Service) GetStreamInfo(user *models_main.User) (string, string, string, string, string, string) {
	streamKey := cache.GenStreamKey(user.Id)
	exis := global2.GetRds().Exists(streamKey).Val()
	var stream string
	var err error
	if exis == 0 {
		stream = util.I2S(user.Id) + "_" + util.I2S(int(time.Now().Unix()))
		err = global2.GetRds().Set(streamKey, stream, time.Duration(86400)*time.Second).Err()
		if err != nil {
			global2.Log.Debugf("GetStreamInfo err:%v", err)
		}
	} else {
		stream, err = global2.GetRds().Get(streamKey).Result()
		if err != nil {
			global2.Log.Debugf("GetStreamInfo err:%v", err)
		}
	}
	push := tx.CreateLiveUrl("", stream, 1)
	anchorSecret := push[2]
	pushUrl := push[0]
	webRtcPushUrl := push[1] + anchorSecret + "&stopstream_api=https://webrtcpush.myqcloud.com/webrtc/v1/stopstream"
	pullUrl := tx.CreateLiveUrl(tx.HostHLS, stream, 0)[0]
	flvPushUrl := tx.CreateLiveUrl(tx.HostHLS, stream, 0)[0]
	return stream, pushUrl, pullUrl, anchorSecret, webRtcPushUrl, flvPushUrl
}

func (s *Service) StartLive(r *api_models.StartLiveReq, user *models_main.User, livePlugins *api_models.LivePlugins) (*models_live.Live, error) {
	fName := "StartLive"

	if user.Id <= 0 {
		return nil, errors.New("主播数据异常")
	}

	var sportId int
	orientation := 1
	if len(r.Pwd) > 0 {
		r.Pwd = encrypt.Md5(r.Pwd)
	}
	live, err := db_live.GetLiveParam(map[string]interface{}{"anchorid": user.Id})
	if err != nil {
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}

	if live.Status == 1 {
		return nil, errors.New("已有开播,请先结束")
	} else if live.Anchorid > 0 {
		err = db_live.DeleteLiveByTransactions(s.DB, live)
		global2.Log.Error("delete models_live  error: ", err)
	}
	stream, err := global2.GetRds().Get(cache.GenStreamKey(user.Id)).Result()
	if err != nil {
		return nil, errors.New("请重新开播")
	}
	if r.MatchId > 0 {
		matchs, err := db_sports.GetMatchsInfoHasStream(r.MatchId)
		if err != nil {
			return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
		}
		if matchs.MatchId <= 0 {
			return nil, errors.New("当前赛事直播信息不存在请选择其它赛事")
		}
		switch matchs.SportId {
		case 1:
			r.SportType = "football"
			sportId = 1
		case 2:
			r.SportType = "basket"
			sportId = 2
		case 3:
			r.SportType = "tennis"
			sportId = 3
		}
	}
	recWeight := user.RecWeight
	recModel := new(models_main.UserRec)
	err = global2.GetR().Model(models_main.UserRec{}).Where("uid = ? AND start_status=1 AND start_time > ?", user.Id, time.Now()).Find(recModel).Error
	if err != nil && recModel.Uid > 0 {
		recWeight = int(recModel.RecWeight) + 100000 //确保后台推荐的权重更大
	}
	liveStatus := livePlugins.LiveStatus
	var pullUrl, pullFlvUrl string
	if livePlugins.PullUrl == "" {
		pullUrl = tx.CreateLiveUrl(tx.HostFLV, stream, 0)[0]
		pullFlvUrl = tx.CreateLiveUrl(tx.HostFLV, stream, 0)[0]
	} else {
		pullUrl = livePlugins.PullUrl
		pullFlvUrl = livePlugins.PullFlvUrl
	}

	configPri := cache.GetConfigPri()
	var heat int
	if configPri.BaseheatsStart >= configPri.BaseheatsEnd {
		heat = configPri.BaseheatsStart
	} else {
		heat = string2.RandInt(configPri.BaseheatsStart, configPri.BaseheatsEnd)
	}
	liveid := int64(util.S2I(util.I2S(int(time.Now().Unix())) + util.Int2SubStr(user.Id, 4, 4)))

	// 标签热度
	var tag models_main.ConfigTag
	if user.Tags > 0 {
		err = global.GetR().Model(models_main.ConfigTag{}).Where("id=?", user.Tags).First(&tag).Error
		if err != nil {
			global2.Log.Debugf("StartLive SetTagBaseHots ERR: %+v", err)
		}
		if tag.Id > 0 {
			if tag.Type == 1 {
				heat = heat + tag.Point
			} else {
				heat = heat - tag.Point
			}
		}

		if heat < 1 {
			heat = 0
		}
	}

	live = &models_live.Live{
		Anchorid:          user.Id,
		Liveid:            liveid,
		Title:             r.Title,
		Thumb:             r.Thumb,
		Stream:            stream,
		StartStamp:        time.Now().Unix(),
		Categoryid:        r.Cateid,
		Orientation:       orientation,
		PullUrl:           pullUrl,
		PullFlvUrl:        pullFlvUrl,
		RecWeight:         recWeight,
		RoomType:          r.RoomType,
		Price:             r.Price,
		Password:          r.Pwd,
		Status:            liveStatus,
		Matchid:           r.MatchId,
		SportType:         r.SportType,
		ThumbY:            r.ThumbY,
		AnchorName:        user.NickName,
		AnchorAvatar:      user.Avatar,
		Category:          r.Category,
		Hot:               heat,
		BaseHot:           heat,
		Platform:          int32(livePlugins.Platform),
		Ticket:            r.Ticket,
		WatermarkPosition: "r",
		PlayFromSource:    livePlugins.PlayFromSource,
		ReplyVideo:        livePlugins.ReplyVideo,
		Tag:               int64(user.Tags),
		SkipTime:          livePlugins.SkipTime,
		EnableWebrtc:      int32(r.EnableWebrtc),
	}
	live.Type = util.GetLiveType(pullUrl)

	session := global2.Live().Begin()
	defer func() {
		if r := recover(); r != nil {
			session.Rollback()
		}
	}()

	if err := session.Error; err != nil {
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}

	if err = db_live.InsertLive(session, live); err != nil {
		session.Rollback()
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}

	if err := session.Commit().Error; err != nil {
		session.Rollback()
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}

	//给直播间接口调用
	anchorLive := &cache.AnchorLive{
		Anchorid:     live.Anchorid,
		Liveid:       live.Liveid,
		Status:       cache.RoomStatus(live.Status),
		Start:        time.Now().Format("2006-01-02 15:04:05"),
		Matchid:      live.Matchid,
		Nickname:     live.AnchorName,
		Heat:         live.Hot,
		OfficialType: 0,
	}
	cache.BeforeDelLiveRoomCache(live.Anchorid)
	err = cache.SetAnchorLive(live.Anchorid, anchorLive)
	if err != nil {
		global2.Log.Debugf("StartLive SetAnchorLive ERR: %+v", err)
	}

	var arg imsdk.OfflineArgs
	var condition imsdk.Condition
	arg.RoomId = 2
	arg.SportType = int32(sportId)
	arg.RoomId = int32(live.Anchorid)
	condition.TagsAnd = []string{strconv.Itoa(live.Anchorid), imsdk.NOTIFY_START_LIVE}

	if live.Platform != 3 {
		fansUids := cache.GetUserAttentAnchorIds(live.Anchorid)
		global2.Log.Debugf("直播开始，关注用户%+v", fansUids)
		if len(fansUids) > 0 {
			var pushUids []string
			for _, uid := range fansUids {
				pushUids = append(pushUids, cast.ToString(uid))
			}
			c := imsdk.OfflineMsgNats{
				OfflineArgs: arg,
				Condition:   condition,
				User:        pushUids,
				Title:       "开播提醒",
				Content:     fmt.Sprintf("您关注的(%s)开播了,赶紧来围观..", live.AnchorName),
			}

			err = global2.PublisherTxChat.BroadCast(publisher2.AllMemberPush, c)
			if err != nil {
				global2.Log.Errorf("NATS PublishAsync StartLive ERR: %+v", err)
				return nil, err
			}
		}
	}

	go CallbackLivePush(&LivePushData{AnchorName: user.NickName, Title: live.Title, Thumb: live.Thumb})

	return live, nil
}

func (s *Service) GetLiveList(r *api_models.GetLiveListReq) (*[]*api_models.LiveInfo, error) {
	fName := "GetLiveList"
	condition := make(map[string]interface{})
	condition["status"] = 1
	//condition["category"] = r.Categoryid
	order := "hot desc"
	if r.Categoryid != 0 {
		if r.Categoryid != 2 {
			r.Categoryid = 1
		}
		condition["category"] = r.Categoryid
	}

	lives, err := db_live.GetLivesParam(condition, order, r.Page, r.Size)
	if err != nil {
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}
	var liveListResp []*api_models.LiveInfo

	mGuess, _ := cache.GetGuessList()
	for _, v := range lives {
		iGuessTag := 0
		if mGuess != nil { //不为空
			if liveId, ok := (*mGuess)[v.Anchorid]; ok && (liveId == v.Liveid) {
				iGuessTag = 1 //如果存在，就标1
			}
		}
		liveListResp = append(liveListResp, &api_models.LiveInfo{
			GuessTag: iGuessTag, //竞猜标签
			Title:    v.Title, Thumb: v.Thumb, AnchorAvatar: v.AnchorAvatar, AnchorName: v.AnchorName,
			Categoryid: uint32(v.Category), Hot: uint32(v.Hot), Anchorid: uint32(v.Anchorid)})

	}
	resp := make([]*api_models.LiveAnchor, 0)
	err = copier.Copy(&resp, &lives)
	if err != nil {
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}
	tmpIds := make([]int, 0)
	for i := range lives {
		tmpIds = append(tmpIds, lives[i].Anchorid)
	}
	return &liveListResp, nil
}

func (s *Service) GetLivePageData(r *api_models.GetLivePageDataReq, clientIP string) (*api_models.GetLivePageDataResp, error) {

	fName := "GetLivePageData"
	/*if exist, err := models.GetBloom().Exists(models.ID, cast.ToString(r.Anchorid)); err != nil || !exist {
		return nil, errors.New("主播不存在:" + fmt.Sprint(exist))
	}*/
	anchorData, err := cache.GetUserFullInfo(r.Anchorid) //repository.GetUserParam(map[string]interface{}{"id": r.Anchorid})
	if err != nil {
		return nil, err
	}
	live, err := db_live.GetLiveParam(map[string]interface{}{"anchorid": r.Anchorid})

	live.TicketPaid = 2
	ticketInLiveFlag, err := db_live.TicketINLive(r.Uid, live.Liveid, cache.GetConfigPub().TicketSwitch)
	if err != nil {
		return nil, err
	}

	if ticketInLiveFlag {
		live.TicketPaid = 1
	}

	// 判断门票入场信息
	// if models_live.Ticket > 0 {
	// 	if r.Uid == 0 {
	// 		return nil, errors.New("本场直播需要门票观看,请登录后继续")
	// 	}
	// 	checked := repository.TicketINLive(r.Uid, models_live.Liveid)
	// 	if !checked {
	// 		var ticket = &models.LiveTicketLog{
	// 			UID:       r.Uid,
	// 			AnchorID:  r.Anchorid,
	// 			Liveid:    models_live.Liveid,
	// 			Ticket:    models_live.Ticket,
	// 			CreatedAt: timehelper.Now().Unix(),
	// 		}

	// 		err = repository.TicketLog(ticket)
	// 		if err != nil {
	// 			return nil, err
	// 		}
	// 	}

	// }

	var userArea *api_models.GetIpSearchResp
	// 后台修改主播资料能及时在直播间生效
	if live.Anchorid > 0 {
		live.AnchorAvatar = anchorData.Avatar
		live.AnchorName = anchorData.NickName
		if global2.Config.GetInt("area_check") == 1 {
			userArea, _ = mysql.GetIpToArea(clientIP)
			// 检查屏蔽库是否返回动画直播
			live = db_live.Blocked(userArea, live)
		}
		if live.PlayFromSource == 1 {
			if live.Category != tx.RePlayType {
				live.Platform = 0 //客户端判断该值为3时，切换点播播放器
			}
			live.PullFlvUrl2 = live.PullUrl
			live.PullRtmpUrl = live.PullUrl
			live.PullM3u8Url = live.PullUrl
		} else {
			live.PullUrl = tx.CreateLiveUrl(tx.HostFLV, live.Stream, 0)[0]
			live.PullFlvUrl = tx.CreateLiveUrl(tx.HostFLV, live.Stream, 0)[0]
			live.PullFlvUrl2 = tx.CreateLiveUrl(tx.HostFLV, live.Stream+global.Config.GetString("livespeed1"), 0)[0]
			live.PullRtmpUrl = tx.CreateLiveUrl(tx.HostRTMP, live.Stream, 0)[0] //---取rtmp的源
			live.PullM3u8Url = tx.CreateLiveUrl(tx.HostHLS, live.Stream, 0)[0]  //---取m3u8的源
		}
	}

	//global2.Log.Debugf("GetLivePageData detail :%+v", live)

	if err != nil {
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}
	param := map[string]interface{}{
		"id": live.Categoryid,
	}
	cates, err := db_live.GetLiveCategoryParam(param, "")
	cateName := ""
	if err == nil && len(cates) > 0 {
		cateName = cates[0].Title
	}
	//anchor, err := s.GetAnchor(r.Anchorid)
	//if err != nil {
	//	return nil, fmt.Errorf("%s%s", common.Convert(fName), err.Error())
	//}
	anchor := new(api_models.Anchor)
	anchor.NickName = anchorData.NickName
	anchor.Id = r.Anchorid
	anchor.Avatar = anchorData.Avatar
	anchor.FansNum = cache.GetFansCount(r.Anchorid)

	if r.Uid > 0 {
		attentedIds := cache.GetUserAttentAnchorIds(r.Uid)
		global2.Log.Debugf("attentedIds:%v\n", attentedIds)
		if string2.IntIsInSlice(attentedIds, r.Anchorid) {
			anchor.Isattent = 1
		}

		// user, err := mysql.GetUserParam(map[string]interface{}{"id": r.Uid})

		// if err == nil {
		// 	// 渠道统计
		// 	if user.RegSourceId != "" {
		// 		buff, err := json.Marshal(user)
		// 		if err != nil {
		// 			global.Log.Errorf("set source stat error[%+v]: ", err)
		// 		} else {
		// 			_, err := global.GetNats().PublishAsync(util.STAT_USER_ACTIVE_CHANNEL, buff, nil)
		// 			if err != nil {
		// 				global.Log.Errorf("Nats publish chanel %s", err.Error())
		// 			}
		// 		}

		// 		err = global2.PublishChannelStats.Send(publisher2.ActiveChannelStats, user)
		// 		if err != nil {
		// 			global2.Log.Errorf("Nats publish chanel %s", err.Error())
		// 		}
		// 	}
		// }
	}
	recLives := make([]*api_models.LiveResp, 0)
	if live.Anchorid <= 0 {
		randLives, err := db_live.GetLivesParam(map[string]interface{}{"status": 1}, "RAND()", 1, 3)
		if err != nil {
			return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
		}
		err = copier.Copy(&recLives, &randLives)
		if err != nil {
			return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
		}
		ids := make([]int, 0)
		for idx := range randLives {
			ids = append(ids, randLives[idx].Anchorid)
		}

		anchorMap, err := s.GetAnchors(0, ids, "", 0, 0)
		if err != nil {
			return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
		}

		for idx := range recLives {
			if _, ok := anchorMap[recLives[idx].Anchorid]; ok {
				recLives[idx].Anchor = anchorMap[recLives[idx].Anchorid]
			}
		}
	}
	var liveNotice = models_live.LiveNotice{}
	err = global2.Live().Model(&models_live.LiveNotice{}).
		Where("anchorid =? or anchorid =0", r.Anchorid).Order("id DESC").First(&liveNotice).Error
	if err != nil {
		global2.Log.Warnf("%s%s", util.Convert(fName), err.Error())
	}

	// 是否开启竞猜
	var _guess int
	err = global2.Guess().Model(&models_guess.Guess{}).Where("live_id=?", live.Liveid).Pluck("id", &_guess).Error
	if err != nil {
		global2.Log.Warnf("%s%s", util.Convert(fName), err.Error())
	}
	if _guess > 0 {
		global.GetRds().Set(_const.ACTION_GUESS_TO_LOGIN+r.DeviceInfo, live.Liveid, 3600*time.Second)
	}
	// ip to 16hex
	return &api_models.GetLivePageDataResp{
		Live:            live,
		CateName:        cateName,
		Anchor:          anchor,
		RecLives:        recLives,
		FansCount:       anchor.FansNum,
		LiveNotice:      liveNotice.Content,
		UserLocal:       fmt.Sprintf("%v", ip2int(net.ParseIP(clientIP))),
		Math:            getMatch(live.Matchid),
		LocalDb:         fmt.Sprintf("userArea:%+v, ip 16hex:%v", userArea, hex.EncodeToString([]byte(clientIP))),
		Advertising:     anchorData.Advertising, //广告词
		ContactName1:    liveNotice.ContactName1,
		ContactAddress1: liveNotice.ContactAddress1,
		ContactName2:    liveNotice.ContactName2,
		ContactAddress2: liveNotice.ContactAddress2,
		ContactName3:    liveNotice.ContactName3,
		ContactAddress3: liveNotice.ContactAddress3,
		MainContent:     liveNotice.MainContent,
	}, nil
}

func ip2int(ip net.IP) uint32 {
	if len(ip) == 16 {
		return binary.BigEndian.Uint32(ip[12:16])
	}
	return binary.BigEndian.Uint32(ip)
}

/**
 * @Author admin
 * @Description //查询赛事Id
 * @version: v1.8.0
 * @Date 14:18 2022/7/30
 * @Param
 * @return
 **/
func getMatch(matchId int) *pt.Match {
	if matchId > 0 {
		match := db_sports.GetMatchById(matchId)
		if match.MatchId > 0 {
			return match
		}
	}
	return nil

}

func (s *Service) HotRecommendLive(categoryid int) ([]*api_models.HotRecommendLive, error) {
	fName := "HotRecommendLive"
	param := map[string]interface{}{"status": 1}
	if categoryid != 0 {
		param["categoryid"] = categoryid
	}
	lives, err := db_live.GetLivesParam(param, "hot desc", 1, 10)
	if err != nil {
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}

	//获取直播主播的信息
	tmpIds := make([]int, 0)
	for i := range lives {
		tmpIds = append(tmpIds, lives[i].Anchorid)
	}

	anchorMap, err := s.GetAnchors(0, tmpIds, "", 0, 0)
	if err != nil {
		return nil, fmt.Errorf("%s%s", util.Convert(fName), err.Error())
	}

	resp := make([]*api_models.HotRecommendLive, 0)

	for _, i := range lives {
		tmp := &api_models.HotRecommendLive{}
		tmp.Hot = i.Hot
		tmp.Status = i.Status
		tmp.Categoryid = i.Categoryid
		tmp.Title = i.Title
		tmp.Matchid = i.Matchid
		tmp.Liveid = i.Liveid
		tmp.PullUrl = i.PullUrl
		tmp.Thumb = i.Thumb
		tmp.ThumbY = i.ThumbY
		tmp.SportType = i.SportType
		tmp.Stream = i.Stream
		tmp.Anchorid = i.Anchorid
		if _, ok := anchorMap[tmp.Anchorid]; ok {
			tmp.Anchor = anchorMap[tmp.Anchorid]
		}
		resp = append(resp, tmp)
	}
	return resp, nil
}

func (s *Service) LiveCallback(r *api_models.LiveCallbackReq) error {
	live, err := db_live.GetLiveParam(map[string]interface{}{"stream": r.StreamId})
	if err != nil {
		return err
	}
	//主播未开播时
	if live.Anchorid <= 0 && r.EventType == 1 {
		params, _ := json.Marshal(r)
		var lcbs = &models_live.LiveCallbackStore{
			StreamID:  r.StreamId,
			Status:    int(cache.StartLive),
			CreatedAt: time.Now(),
			Params:    string(params),
		}

		err = global2.Live().Create(&lcbs).Error

		if err != nil {
			return err
		}
		return nil
	}

	if r.EventType == 1 && live.Status == int(cache.PauseLive) { //直播推流
		live.Status = int(cache.StartLive)
		err = db_live.UpdateLiveByTransactions(nil, []string{"status"}, live)

		if err != nil {
			global2.Log.Error(err)
			return err
		}
		anchorLive, _ := cache.GetAnchorLive(live.Anchorid)

		l := &cache.AnchorLive{
			Anchorid: live.Anchorid,
			Liveid:   live.Liveid,
			Start:    timehelper.Stamp2Str(live.StartStamp),
			Status:   cache.StartLive,
			Matchid:  live.Matchid,
			Nickname: anchorLive.Nickname,
			Heat:     anchorLive.Heat,
		}
		cache.SetAnchorLive(live.Anchorid, l)
		//s.DelCallBackLiveData(live.Stream)
		return nil
	}
	//直播斷流五分鐘進行查看是否還在沒如果不在入庫
	if r.EventType == 0 && live.Anchorid > 0 {
		s.SetCallBackLiveData(r.StreamId, live.Anchorid)
	}
	return nil
}

// 录播回调主播查询
func (s *Service) AnchorLiveHis(stream string) (int, string, string, error) {
	var live models_live.LiveHistory
	callback := strings.Split(stream, "_")
	if len(callback) < 2 {
		return 0, "", "", errors.New("record callback error")
	}
	err := global2.Live().Model(&models_live.LiveHistory{}).Where("stream=?", callback[0]+"_"+callback[1]).First(&live).Error
	if err != nil {
		return 0, "", "", err
	}

	return live.Anchorid, live.Title, callback[0] + "_" + callback[1], nil
}

// 录播回调写入视频库
func (s *Service) Record2Video(record *models_live.Video) error {
	return global2.GetX().Create(&record).Error
}

// 活动
func (s *Service) Activity(anchorID int) (*models_main.ActivityBarrage, error) {
	// 从 Redis 检查主播的活动状态
	activityKey := cache.GenActivityEnabledKey(anchorID)
	enabled, err := global.GetRds().Get(activityKey).Int()
	if err != nil {
		// 只有当Redis中没有值时，才查询数据库
		var count int64
		err := global2.Live().Model(&models_live.Live{}).Where("anchorid=? and activity_toggle", anchorID).Count(&count).Error
		if err != nil || count == 0 {
			// 即使查询失败或没有记录，也设置Redis缓存，避免重复查询数据库
			global.GetRds().Set(activityKey, 0, 24*time.Hour)
			return nil, err
		}
		// 如果数据库查询成功，设置Redis
		enabled = 1
		global.GetRds().Set(activityKey, enabled, 24*time.Hour)
	} else if enabled != 1 {
		// Redis中有值，但值不为1（活动未启用）
		// 保持与数据库查询不到记录时相同的返回格式
		return nil, nil
	}

	// 首先获取活动ID
	var abid int
	abidKey := cache.GenActivityIDKey(anchorID)
	abid, err = global.GetRds().Get(abidKey).Int()
	if err != nil {
		// 只有当Redis中没有值时，才查询数据库
		err = global.GetR().Model(&models_main.ActivityBarrage{}).Where("status=1").Pluck("id", &abid).Error
		if err != nil {
			global.Log.Infof("Activity ID Error: %s", err.Error())
			// 即使查询失败，也设置Redis缓存，避免重复查询数据库
			global.GetRds().Set(abidKey, -1, 24*time.Hour)
			return nil, nil
		}
		// 缓存活动ID
		global.GetRds().Set(abidKey, abid, 24*time.Hour)
	} else if abid <= 0 {
		// Redis中有值，但值小于等于0（无效活动ID）
		return nil, nil
	}

	// 从 Redis 获取活动数据
	activityDataKey := cache.GenActivityDataKey(abid)
	activityData, err := global.GetRds().HGetAll(activityDataKey).Result()

	var ab models_main.ActivityBarrage

	if err != nil || len(activityData) == 0 {
		// 只有当Redis中没有值时，才查询数据库
		err = global2.GetX().Model(&ab).Where("status=1").First(&ab).Error
		if err != nil {
			// 即使查询失败，也设置Redis缓存，避免重复查询数据库
			emptyData := map[string]interface{}{"empty": "1"}
			global.GetRds().HMSet(activityDataKey, emptyData)
			global.GetRds().Expire(activityDataKey, 24*time.Hour)
			return nil, nil
		}

		// 如果数据库查询成功，设置Redis
		activityMap := map[string]interface{}{
			"id":           ab.ID,
			"link":         ab.Link,
			"icon":         ab.Icon,
			"desc":         ab.Desc,
			"status":       ab.Status,
			"barrages":     ab.Barrages,
			"toggle_pop":   ab.TogglePop,
			"toggle_enter": ab.ToggleEnter,
			"login":        ab.Login,
		}
		global.GetRds().HMSet(activityDataKey, activityMap)
		global.GetRds().Expire(activityDataKey, 24*time.Hour)
	} else if _, exists := activityData["empty"]; exists {
		// Redis中有标记表示数据为空
		return nil, nil
	} else {
		// 从 Redis 解析数据
		ab.ID = cast.ToInt(activityData["id"])
		ab.Link = activityData["link"]
		ab.Icon = activityData["icon"]
		ab.Desc = activityData["desc"]
		ab.Status = cast.ToInt32(activityData["status"])
		ab.Barrages = activityData["barrages"]
		ab.TogglePop = cast.ToInt32(activityData["toggle_pop"])
		ab.ToggleEnter = cast.ToInt32(activityData["toggle_enter"])
		ab.Login = cast.ToInt32(activityData["login"])
	}

	// 从 Redis 获取直播 ID
	liveIDKey := cache.GenActivityLiveIDKey(anchorID)
	liveID, err := global.GetRds().Get(liveIDKey).Int64()

	if err != nil {
		// 只有当Redis中没有值时，才查询数据库
		err = global.Live().Model(&models_live.Live{}).Where("anchorid=?", anchorID).Pluck("liveid", &liveID).Error
		if err != nil {
			global.Log.Infof("Activity Stat Error: %s", err.Error())
			// 即使查询失败，也设置Redis缓存，避免重复查询数据库
			global.GetRds().Set(liveIDKey, -1, 24*time.Hour)
		} else {
			// 如果数据库查询成功，设置Redis
			global.GetRds().Set(liveIDKey, liveID, 24*time.Hour)
		}
	} else if liveID < 0 {
		// Redis中标记了无效的liveID，使用默认值0
		liveID = 0
	}

	// 从 Redis 获取参与统计
	statKey := cache.GenActivityStatKey(anchorID*100, ab.ID, liveID)
	//log.Println("riskFilter=====>key:", statKey)
	rets, _ := global.GetRds().HGetAll(statKey).Result()

	ab.Times = len(rets)
	ab.RoomID = anchorID * 100
	ab.AnchorID = anchorID

	return &ab, nil
}

// ES记录观众渠道来源
const ES_USER_CHANNEL = "es_live_user_channel"

func UserChannel2ES(liveid string, device string, channel string) {
	if device == "" || liveid == "" || channel == "" {
		return
	}
	var user = struct {
		Liveid  string `json:"liveid"`
		Device  string `json:"device"`
		Channel string `json:"channel"`
	}{
		Liveid:  liveid,
		Device:  device,
		Channel: channel,
	}

	// if userChannel2ESCheck(liveid, device) {
	// 	return
	// }

	id := int64(snowflake.ID())
	jsonData, _ := json.Marshal(&user)
	es := global.GetEs().Index().Index(ES_USER_CHANNEL).Id(cast.ToString(id))
	es.BodyString(string(jsonData))
	res, err := es.Do(context.Background())
	log.Println("res:", res)
	log.Println("err:", err)
}

func userChannel2ESCheck(liveid string, device string) bool {

	log.Println("liveid:", liveid)
	log.Println("device:", device)

	// termQuery := elastic.NewTermQuery("liveid", liveid)
	res, err := global.GetEs().Search().
		Index(ES_USER_CHANNEL).
		// Query(elastic.NewTermQuery("liveid", liveid)).
		Query(elastic.NewTermQuery("device", device)).
		// Query(elastic.NewTermQuery("channel", "ES888")).
		Do(context.TODO())
	log.Println("res:", res.Hits.Hits)
	log.Println("err:", err)

	log.Println("es.Hits.TotalHits:", res.Hits.TotalHits)
	log.Println("es.Hits.TotalHits.Value:", res.Hits.TotalHits.Value)
	return res.Hits.TotalHits.Value > 0
}
